package com.ds.lens.flink.monitor.common;

import com.ds.lens.data.common.client.LensMonitorClient;
import com.ds.lens.data.common.client.LensResponseData;
import com.ds.lens.data.common.client.vo.LensMonitorDetailsOutVO;
import com.ds.lens.data.common.constant.AggregatorEnum;
import com.ds.lens.data.common.constant.CompareOperatorEnum;
import com.ds.lens.data.common.constant.DetectionTypeEnum;
import com.ds.lens.data.common.constant.MonitorStateEnum;
import com.ds.lens.data.common.document.dto.MonitorKey;
import com.ds.lens.flink.monitor.common.bo.EventNotificationDefinition;
import com.ds.lens.flink.monitor.MonitorJobContext;
import com.ds.lens.flink.monitor.common.bo.MonitorDefinition;
import com.ds.lens.flink.monitor.common.bo.MonitorState;
import com.ds.lens.flink.monitor.monitorjob.function.detector.*;

import java.util.Map;

public class MonitorDefinitionLoadFunction implements LoadFunction<Long, MonitorDefinition> {
    @Override
    public MonitorDefinition load(Long monitorId) throws Exception {
        LensMonitorClient monitorClient = MonitorJobContext.getInstance().getMonitorClient();
        try {
            LensResponseData<LensMonitorDetailsOutVO> resp = monitorClient.getMonitorDefinition(monitorId);
            LensMonitorDetailsOutVO respVO = resp.getData();
            if (respVO == null) {
                return null;
            }
            return buildMonitorDefinition(respVO);
        } catch (Exception e) {
            e.printStackTrace();
            return null;
        }
    }

    private MonitorDefinition buildMonitorDefinition(LensMonitorDetailsOutVO respVO) {
        MonitorState monitorState = MonitorState.builder()
                .key(new MonitorKey(respVO.getDefine().getMonitorId(), respVO.getVersion()))
                .isDelete(respVO.getIsDelete())
                .runStatus(respVO.getRunStatus())
                .applicationId(respVO.getApplicationId())
                .modifyTime(respVO.getModifiedTime())
                .build();

        return MonitorDefinition.builder()
                .integrationId(respVO.getIntegrationId())
                .monitorName(respVO.getName())
                .state(MonitorStateEnum.valueOf(respVO.getState()))
                .groupBys(respVO.getDefine().getMetricBy())
                .tags(respVO.getTags())
                .scope(respVO.getDefine().getScope())
                .filters(respVO.getDefine().getMetricFilter())
                .intervalTime(respVO.getDefine().getWindowTime())
                .metricAggr(AggregatorEnum.valueOf(respVO.getDefine().getMetricAggr()))
                .metricName(respVO.getDefine().getMetricName())
                .monitorState(monitorState)
                .retiveDuration(respVO.getDefine().getDuring())
                .retriveDelayTime(respVO.getDefine().getDelayTime())
                .windowTime(respVO.getDefine().getWindowTime())
                .windowType(respVO.getDefine().getWindowType())
                .detector(buildMonitorDetector(respVO))
                .notification(buildEventNotification(respVO))
                .versionTime(respVO.getVersionTime())
                .levelThresholdMap(respVO.getDefine().getThresholdLevel())
                .detectionMethod(respVO.getDetectionMethod())
                .build();
    }

    private MonitorEventDetector buildMonitorDetector(LensMonitorDetailsOutVO respVO) {
        MonitorEventDetectorChain chain = new MonitorEventDetectorChain();
        chain.addDetector(new NoDataDetector());
        Map<String, Object> levelMap = respVO.getDefine().getThresholdLevel();
        if (levelMap != null) {
            for (String key : levelMap.keySet()) {
                double threshold = Double.parseDouble(levelMap.get(key).toString());
                MonitorEventDetector detector;
                if (respVO.getDetectionMethod() == null || respVO.getDetectionMethod().equals(Integer.valueOf(DetectionTypeEnum.STATIC_STATE.getCode()))) {
                    detector = ThresholdDetector.builder()
                            .state(MonitorStateEnum.valueOf(Integer.valueOf(key)))
                            .atLeastTimes(respVO.getDefine().getThresholdNumber())
                            .operator(CompareOperatorEnum.valueOf(respVO.getDefine().getOperator()))
                            .threshold(threshold)
                            .aggrEnum(AggregatorEnum.valueOf(respVO.getDefine().getMetricAggr()))
                            .metricName(respVO.getDefine().getMetricName())
                            .build();
                } else {
                    detector = DynamicDetector.builder()
                            .state(MonitorStateEnum.valueOf(Integer.valueOf(key)))
                            .atLeastTimes(respVO.getDefine().getThresholdNumber())
                            .operator(CompareOperatorEnum.valueOf(respVO.getDefine().getOperator()))
                            .threshold(threshold)
                            .build();
                }
                chain.addDetector(detector);
            }
        }
        return chain;
    }

    private EventNotificationDefinition buildEventNotification(LensMonitorDetailsOutVO respVO) {
        return EventNotificationDefinition.builder()
                .content(respVO.getNotify().getContent())
                .renotify(respVO.getNotify().getRenotifyTime() > 0)
                .renotifyDuration(respVO.getNotify().getRenotifyTime())
                .notifyNoData(respVO.getDefine().getNotifyDataMissing())
                .notifyNoDataDuration(respVO.getDefine().getWindowTime())
                .isMute(respVO.getMute())
                .muteStartTime(respVO.getMuteStartTime())
                .muteEndTime(respVO.getMuteEndTime())
                .title(respVO.getNotify().getTitle())
                .channel(respVO.getNotify().getChannel())
                .channelContent(respVO.getNotify().getChannelContent())
                .build();
    }
}
